package data

import (
	"database/sql"
	"errors"
	"fmt"
	"strings"

	_ "github.com/go-sql-driver/mysql"

	"mig/pkg/chunk"
	"mig/pkg/types"
)

const (
	subBatchSize = 1000
)

type Mysql struct {
	db        *sql.DB
	cfg       *types.MysqlConfig
	schema    *types.Schema
	algorithm types.Algorithm
}

func (m *Mysql) Execute(sql string) error {
	_, err := m.db.Exec(sql)
	return err
}

func NewMysql(cfg *types.MysqlConfig, schema *types.Schema, algorithm types.Algorithm) *Mysql {
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", cfg.Username, cfg.Password, cfg.Host, cfg.Port, cfg.Database)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		panic(err)
	}
	return &Mysql{db, cfg, schema, algorithm}
}

func (m *Mysql) Identifier() string {
	return fmt.Sprintf("%s_%s", m.cfg.Identifier(), m.schema.Name)
}

func (m *Mysql) ReadChunk(chunkNo, chunkSize int64) (chunk.Chunk, error) {
	offset := chunkNo * chunkSize
	id := m.schema.PrimaryField
	fieldNames := m.schema.FieldNames()
	sql := fmt.Sprintf("SELECT %s FROM %s ORDER BY %s LIMIT %d OFFSET %d", strings.Join(fieldNames, ","),
		m.schema.Name, id, chunkSize, offset)
	fmt.Println(sql)
	rows, err := m.db.Query(sql)
	if err != nil {
		return nil, err
	}

	ck := chunk.NewChunk(m.algorithm)
	var record *types.Record
	for rows.Next() {
		record, err = types.NewRecord(m.schema.Fields)
		if err != nil {
			return nil, err
		}
		err = rows.Scan(record.FieldRefs()...)
		if err != nil {
			return nil, err
		}
		ck.AddRecord(record)
	}
	ck.Finalize()
	return ck, nil
}

func (m *Mysql) Count() (int64, error) {
	sql := fmt.Sprintf("SELECT COUNT(*) FROM %s", m.schema.Name)
	var count int64
	err := m.db.QueryRow(sql).Scan(&count)
	if err != nil {
		return 0, err
	}
	return count, nil
}

func (m *Mysql) WriteChunk(chunk chunk.Chunk) error {
	if chunk == nil {
		return nil
	}

	records := chunk.Records()
	if len(records) == 0 {
		return nil
	}

	nFields := m.schema.NumFields()
	if nFields == 0 {
		return errors.New("schema has no fields")
	}

	placeholders := make([]string, 0, nFields)
	for range m.schema.Fields {
		placeholders = append(placeholders, "?")
	}
	singleRowPlaceholders := "(" + strings.Join(placeholders, ", ") + ")"

	updateFields := make([]string, 0, nFields)
	for _, field := range m.schema.FieldNames() {
		if field == m.schema.PrimaryField {
			continue
		}
		updateFields = append(updateFields, fmt.Sprintf("%s = VALUES(%s)", field, field))
	}
	onDuplicateClause := "ON DUPLICATE KEY UPDATE " + strings.Join(updateFields, ", ")

	fieldList := strings.Join(m.schema.FieldNames(), ", ")

	for start := 0; start < len(records); start += subBatchSize {
		end := start + subBatchSize
		if end > len(records) {
			end = len(records)
		}

		batchRecords := records[start:end]
		allPlaceholders := make([]string, 0, len(batchRecords))
		allValues := make([]interface{}, 0, len(batchRecords)*nFields)

		for i := 0; i < len(batchRecords); i++ {
			allPlaceholders = append(allPlaceholders, singleRowPlaceholders)
			allValues = append(allValues, batchRecords[i].FieldRefs()...)
		}

		insertSQL := fmt.Sprintf("INSERT INTO %s (%s) VALUES %s %s",
			m.schema.Name, fieldList, strings.Join(allPlaceholders, ", "), onDuplicateClause)

		_, err := m.db.Exec(insertSQL, allValues...)
		if err != nil {
			return fmt.Errorf("insert failed: %w", err)
		}
	}

	return nil
}

//func (m *Mysql) WriteChunk(chunk chunk.Chunk) error {
//	if chunk == nil {
//		return nil
//	}
//
//	placeholders := make([]string, 0, m.schema.NumFields())
//	for range m.schema.Fields {
//		placeholders = append(placeholders, "?")
//	}
//	singleRowPlaceholders := "(" + strings.Join(placeholders, ", ") + ")"
//
//	records := chunk.Records()
//	nRecords := len(records)
//	nFields := m.schema.NumFields()
//	allPlaceholders := make([]string, 0, nRecords)
//	allValues := make([]interface{}, 0, nRecords*nFields)
//	for i := 0; i < nRecords; i++ {
//		allPlaceholders = append(allPlaceholders, singleRowPlaceholders)
//		allValues = append(allValues, records[i].FieldRefs()...)
//	}
//	fieldList := strings.Join(m.schema.FieldNames(), ", ")
//	insertSQL := fmt.Sprintf("INSERT INTO %s (%s) VALUES %s", m.schema.Name, fieldList, strings.Join(allPlaceholders, ", "))
//	_, err := m.db.Exec(insertSQL, allValues...)
//	return err
//}
